package com.chencong.udf;

import com.chencong.bean.UserOrderCount;
import com.chencong.utils.GetMysqlConnect;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @Author chencong
 * @Description 读取mysql函数
 * @Date 7:04 下午 2021/8/17
 * @Param
 **/
public class SourceFromMysqlFunc extends RichSourceFunction<UserOrderCount> {
    PreparedStatement ps;

    private Connection connect;

    /**
     * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。
     *
     * @param
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connect = GetMysqlConnect.getConnect();
        String sql = "select * from user_order_count;";
        ps = this.connect.prepareStatement(sql);
    }

    /**
     * 程序执行完毕就可以进行，关闭连接和释放资源的动作了
     *
     * @throws Exception
     */

    @Override
    public void close() throws Exception {
        super.close();
        if (connect != null) { //关闭连接和释放资源
            connect.close();
        }

        if (ps != null) {
            ps.close();
        }
    }

    /**
     * 功能描述: 主要逻辑
     * @Param: [ctx]
     * @Return: [org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext]
     * @Author: chencong
     * @Date: 2021/8/17 7:31 下午
     */
    @Override
    public void run(SourceContext<UserOrderCount> ctx) throws Exception {
        ResultSet resultSet = ps.executeQuery();// 执行SQL语句返回结果集

        while (resultSet.next()) {
            UserOrderCount userOrderCount=  new UserOrderCount(
                    resultSet.getString("user_id").trim(),
                    resultSet.getInt("count"));
            ctx.collect(userOrderCount);

        }



    }

    @Override
    public void cancel() {

    }
}
//1、run ： 启动一个 source，即对接一个外部数据源然后 emit 元素形成 stream（大部分情况下会通过在该方法里运行一个 while 循环的形式来产生 stream）。
//2、cancel ： 取消一个 source，也即将 run 中的循环 emit 元素的行为终止。
//正常情况下，一个 SourceFunction 实现这两个接口方法就可以了。其实这两个接口方法也固定了一种实现模板。
